package com.atguigu.Flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;
import java.util.List;

public class Flink02_OperatorListState {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //配置程序失败恢复策略
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(2)));
        //开启检查点
        env.enableCheckpointing(2000L);
        DataStreamSource<String> ds = env.socketTextStream("hadoop102", 8888);
        ds.map(
                new MyMapFunction()
        ).addSink(
                new SinkFunction<String>() {
                    @Override
                    public void invoke(String value, Context context) throws Exception {
                        System.out.println(value);
                        if(value.contains("s")){
                            throw new RuntimeException("出现异常");
                        }
                    }
                }
        );


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
    public  static  class  MyMapFunction implements MapFunction<String,String>, CheckpointedFunction {
        //自定义原始状态
        //private List<String> words = new ArrayList<>();
        //算子状态
        private ListState<String> listState;

        @Override
        public String map(String s) throws Exception {
            listState.add(s);
            return listState.get().toString();
        }

        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
            System.out.println("aaaaaaaaaaaaaaaaaaaaaaaaaa");
        }

        @Override
        public void initializeState(FunctionInitializationContext functionInitializationContext) throws Exception {
            System.out.println("bbbbbbbbbbbbbbbbbbbbbbbbbb");
            //初始化算子状态
            ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<String>("liststate", Types.STRING);
             listState = functionInitializationContext.getOperatorStateStore().getListState(listStateDescriptor);

        }
    }
}
